package cn.dmrliu.edu.realtime.util;

import cn.dmrliu.edu.realtime.common.EduConfig;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

import java.io.IOException;

public class KafkaUtil {
    private static final String KAFKA_URL = "h102:9092,h103:9092,h104:9092";

    /**
     * 创建kafkasource
     *
     * @param topic 消费主题
     * @param group 消费组
     */
    public static KafkaSource<String> getKafkaSource(String topic, String group) {
        KafkaSource<String> build = KafkaSource.<String>builder()
                // kafka地址
                .setBootstrapServers(KAFKA_URL)
                // kafka 消费主题
                .setTopics(topic)
                // kafka 消费组
                .setGroupId(group)
                // kafka 消费偏移量
                .setStartingOffsets(OffsetsInitializer.latest())
                // kafka 反序列化
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] message) throws IOException {
                        if (message != null) {
                            return new String(message);
                        }
                        return null;
                    }

                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return TypeInformation.of(String.class);
                    }
                })
                .build();
        return build;
    }

    public static KafkaSink<String> getKafkaSink(String topic) {
        KafkaSink<String> sink = KafkaSink.<String>builder()
                .setBootstrapServers(EduConfig.KAFKA_SERVER)
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.builder()
                                .setTopic(topic)
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()
                )
                .build();
        return sink;
    }

    public static <T> KafkaSink<T> getKafkaSinkBySchema(KafkaRecordSerializationSchema<T> kafkaRecordSerializationSchema) {
        KafkaSink<T> kafkaSink =
                KafkaSink.<T>builder()
                        .setBootstrapServers(EduConfig.KAFKA_SERVER)
                        .setRecordSerializer(kafkaRecordSerializationSchema)
                        .build();
        return kafkaSink;
    }


    // ods 对应的动态表 建表语句
    public static String getEduDbDDL(String group) {
        return "CREATE TABLE edu_db (\n" +
                "  `database` string,\n" +
                "  `table` string,\n" +
                "  `type` string,\n" +
                "  `ts` string,\n" +
                "  `data` MAP<string, string>,\n" +
                "  `old` MAP<string, string>,\n" +
                "  proc_time as proctime()\n" +
                ")"
                + getKafkaDDL(group);
    }

    // kafka 连接器的连接字段
    public static String getKafkaDDL(String group) {
        return "WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'edu_db',\n" +
                "  'properties.bootstrap.servers' = '" + EduConfig.KAFKA_SERVER + "',\n" +
                "  'properties.group.id' = '" + group + "',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json'\n" +
                ")";
    }

    public static String getKafkaDDL(String topic, String group) {
        return "WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '" + topic + "',\n" +
                "  'properties.bootstrap.servers' = '" + EduConfig.KAFKA_SERVER + "',\n" +
                "  'properties.group.id' = '" + group + "',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json'\n" +
                ")";
    }

    // upsert kafka 连接器的连接字段
    public static String getUpsertKafkaDDL(String topic) {
        return "WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = '" + topic + "',\n" +
                "  'properties.bootstrap.servers' = '" + EduConfig.KAFKA_SERVER + "',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ")";
    }
}
